package com.fwmagic.spark.operation

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
  * DataSet:强类型使用
  */
object TypedOperation {

  def main(args: Array[String]): Unit = {
     val spark = SparkSession
       .builder()
       .appName("TypedOperation")
       .master("local[*]")
       .getOrCreate()

    import spark.implicits._

    //读取json文件
    val employee: DataFrame = spark.read.json("/Users/fangwei/learn/mycode/workspace/fwmagic-spark/src/main/resources/employee.json")
    val employee2: DataFrame = spark.read.json("/Users/fangwei/learn/mycode/workspace/fwmagic-spark/src/main/resources/employee2.json")

    //读取json文件
    val department = spark.read.json("/Users/fangwei/learn/mycode/workspace/fwmagic-spark/src/main/resources/department.json")


    //DataFrame to DataSet
    val empDS: Dataset[Employee] = employee.as[Employee]
    val empDS2: Dataset[Employee] = employee2.as[Employee]

    val depDS: Dataset[Department] = department.as[Department]

    //查看分区数
    println(empDS.rdd.partitions.size)
    println(depDS.rdd.partitions.size)

    /**
      * coalesce 和 repartition 操作
      * 都是用来重新丁意思分区的
      * 区别在于：coalese只能用来减少分区，而且可以选择不发生分区
      * repartition可以增加分区，也可以减少分区，必须会发生shuffle，相当是进行了一次重分区操作
      */

    /*val reparDS: Dataset[Employee] = empDS.repartition(7)
    //打印分区数
    println(reparDS.rdd.partitions.size)

    val secondDS: Dataset[Employee] = reparDS.repartition(3)
    println(secondDS.rdd.partitions.size)

    val coalesDS: Dataset[Employee] = secondDS.coalesce(2)
    println(coalesDS.rdd.partitions.size)

    coalesDS.show()*/


    /**
      * distinct和dropDuplicates
      * 都是用来去重的
      * 区别：
      * distinct：是根据每一条内容的完整内容进行对比和去重
      * dropDuplicates：可以根据指定的字段进行去重
      */
      empDS.distinct().show()

      println("==========================")
      empDS.dropDuplicates("name").show()


    /**
      * expect:获取当前dataset中有，但是另外一个dataset中没有的数据
      * filter:根据自己的的逻辑，获取所有返回为true的数据
      * intersect:获取两个数据集的交集(两个集合都有的数据)
      */
      empDS.except(empDS2).show()
      empDS.filter(e => e.age > 40).show()
      empDS.intersect(empDS2).show()

    /**
      * map:将数据集中的每一条数据都做一个映射，返回一条新数据
      * flatMap:数据集中的每一条数据都可以返回多条数据
      * mapPartitions:一次性对一个分区的数据做处理
      */
      empDS.map(e => (e.name,e.age+100)).show()
      println("===============")
      depDS.flatMap(dep => Seq(Department(dep.id+1,dep.name+"_1"),Department(dep.id+2,dep.name+"_2"))).show()

      empDS.mapPartitions(emp => {
        var result = scala.collection.mutable.ArrayBuffer[(String,Double)]()
        while(emp.hasNext){
          val employee: Employee = emp.next()
          result += ((employee.name,employee.salary+0.5))
        }
        result.iterator
      }).show()

    //joinWith
    empDS.joinWith(depDS,empDS("depId") === depDS("id")).show()


   // empDS.randomSplit(Array(2,3,10)).foreach(ds=> ds.show())

    //抽样
    //empDS.sample(false,0.1).show()

    spark.stop()
  }

  case class Employee(name:String,age:Long,depId:Long,gender:String,salary:Double)

  case class Department(id:Long,name:String)

}
